using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Reactive.Linq;
using System.Reactive.Subjects;

namespace Provider
{
    public class RefCountSubscriber<TKey, TValue> : IRefCountSubscriber<TKey, TValue>
        where TKey : IKey
        where TValue : IValue
    {
        private readonly ISubscriberComponent<TKey,TValue> _subscriberComponent;

        private readonly ConcurrentDictionary<TKey, IObservable<KeyValuePair<TKey, TValue>>> _observables = 
            new ConcurrentDictionary<TKey, IObservable<KeyValuePair<TKey, TValue>>>();

        public RefCountSubscriber(ISubscriberComponent<TKey, TValue> subscriberComponent)
        {
            _subscriberComponent = subscriberComponent;
        }

        public IObservable<KeyValuePair<TKey, TValue>> GetUpdates(TKey key)
        {
            IObservable<KeyValuePair<TKey, TValue>> observable;
            if (!_observables.ContainsKey(key))
            {
                observable = Observable
                    .Create<KeyValuePair<TKey, TValue>>((o) =>
                        {
                            _subscriberComponent
                                .Updates
                                .Where(x=> x.Key.Equals(key))
                                .Subscribe(update => o.OnNext(update));
                            _subscriberComponent.Subscribe(key);
                            return () =>
                            {
                                IObservable<KeyValuePair<TKey, TValue>> observableToRemove;
                                _subscriberComponent.UnSubscribe(key);
                                _observables.TryRemove(key, out observableToRemove);
                            };
                        })
                    .Publish()
                    .RefCount();
                _observables.AddOrUpdate(key, observable, (_, entry) => entry);
            }
            else
            {
                observable = _observables[key];
            }

            return observable.StartWith(_subscriberComponent.GetIntialValue(key));
        }

        public IObservable<KeyValuePair<TKey, TValue>> GetUpdates(IEnumerable<TKey> keys)
        {
            return keys
                .Select(GetUpdates)
                .Aggregate((a, b) => a.Merge(b));
        }

        public IEnumerable<TKey> SubscriptionKeys
        {
            get { return _observables.Keys; }
        }

        public void Dispose()
        {
            _subscriberComponent.Dispose();
        }


        public virtual void Initialise(Dictionary<string, string> args)
        {
        }
    }
}